package com.stars.easyms.schedule.factory;

import com.stars.easyms.schedule.exception.DistributedScheduleMQException;
import com.stars.easyms.schedule.mq.MQMessageSender;
import com.stars.easyms.schedule.mq.activemq.ActiveMQMessageSender;
import com.stars.easyms.schedule.mq.rocketmq.RocketMQMessageSender;
import com.stars.easyms.schedule.util.ReflectUtil;
import com.stars.easyms.schedule.util.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * MQ消息发送者工厂类
 *
 * @author guoguifang
 */
public class MQMessageSenderFactory implements MQFactory {

    private static Map<String, MQMessageSender> activeMQMessageSenderMap;

    private static Map<String, MQMessageSender> rocketMQMessageSenderMap;

    /**
     * 获取ActiveMQ消息发送对象
     * @param brokerUrl 连接mq的brokerUrl
     * @param user 连接mq的用户名
     * @param password 连接mq的密码
     * @return MQ消息发送对象
     */
    public static MQMessageSender getActiveMQMessageSender(String brokerUrl, String user, String password) throws DistributedScheduleMQException {
        if (activeMQMessageSenderMap == null) {
            synchronized (MQMessageSenderFactory.class) {
                if (activeMQMessageSenderMap == null) {
                    activeMQMessageSenderMap = new ConcurrentHashMap<>(64);
                }
            }
        }
        String key = StringUtils.join(new Object[]{brokerUrl, user, password}, "-");
        MQMessageSender activeMQMessageSender = activeMQMessageSenderMap.get(key);
        if (activeMQMessageSender == null) {
            if (ReflectUtil.forNameWithNoException(JMS_CONN_FACTORY) == null) {
                throw new DistributedScheduleMQException("Could not find class '" + JMS_CONN_FACTORY + "'!");
            }
            if (ReflectUtil.forNameWithNoException(ACTIVEMQ_CONN_FACTORY) == null) {
                throw new DistributedScheduleMQException("Could not find class '" + ACTIVEMQ_CONN_FACTORY + "'!");
            }
            activeMQMessageSender = new ActiveMQMessageSender(brokerUrl, user, password);
            activeMQMessageSenderMap.put(key, activeMQMessageSender);
        }
        return activeMQMessageSender;
    }

    /**
     * 获取RocketMQ消息发送对象
     * @param nameServer 如果使用的是rocketmq则需要增加nameServer
     * @param groupName 如果使用的是rocketmq则需要增加groupName
     * @return MQ消息发送对象
     */
    public static MQMessageSender getRocketMQMessageSender(String nameServer, String groupName) throws DistributedScheduleMQException {
        if (rocketMQMessageSenderMap == null) {
            synchronized (MQMessageSenderFactory.class) {
                if (rocketMQMessageSenderMap == null) {
                    rocketMQMessageSenderMap = new ConcurrentHashMap<>(64);
                }
            }
        }
        String key = StringUtils.join(new Object[]{nameServer, groupName}, "-");
        MQMessageSender rocketMQMessageSender = rocketMQMessageSenderMap.get(key);
        if (rocketMQMessageSender == null) {
            if (ReflectUtil.forNameWithNoException(ROCKETMQ_PRODUCER) == null) {
                throw new DistributedScheduleMQException("Could not find class '" + ROCKETMQ_PRODUCER + "'!");
            }
            rocketMQMessageSender = new RocketMQMessageSender(nameServer, groupName);
            rocketMQMessageSenderMap.put(key, rocketMQMessageSender);
        }
        return rocketMQMessageSender;
    }
}
